
#include <stdlib.h>
#include <time.h>
#include <stdio.h>
#include <memory.h>
#include <errno.h>
#include <assert.h>

#define LP_CORE

#include "lp_socket.h"
#include "lp_service.h"


#ifndef LP_WINDOWS
    #include <sys/socket.h>
    #include <netinet/in.h>
    #include <netdb.h>
    #include <sys/types.h>
    #include <sys/ioctl.h>
    #include <sys/select.h>
    #include <fcntl.h>
    #include <unistd.h>
    #include <signal.h>
#endif


#define DEFAULT_RECV_SIZE 20480
#define MAX_UDP_LEN 65535

enum {
    SS_CONNECTING,
    SS_CONNECTED,
    SS_BIND,
    SS_LISTEN,
    SS_ACCEPT,
};


struct LPSocketAttacher {
    LPService attach;
    LPSocketPoll* poll;
    char udpbuffer[MAX_UDP_LEN];
    char tcpbuffer[DEFAULT_RECV_SIZE];
};

typedef struct BuffList BuffList;
struct BuffList {
    size_t size;
    LPSession session;
    char* data;
    BuffList* next;
};


typedef struct {
    LPSOCK_FD fd;
    LPSocket so;
    int protocol;
    int family;
    LPSocketAddr addr;
    int status;
    int evt;
    int forward;
    LPSession session;
    LPSocketAttacher* attacher;
    size_t cbufSize;
    BuffList* cbuf;
    BuffList* cbufEnd;
    BuffList* rbuf;
    BuffList* rbufEnd;
    BuffList* sbuf;
    BuffList* sbufEnd;
}Socket;


static struct{
    LPStorage* storage;
}gCtx;



static void
freeSocketMsg(LPMsg* msg) {
    free(msg->data);
}

static void handlePipe(int sig) {
    printf("handlePipe: %d\n", sig);
}

static void
respSocketMsg(Socket* s, int cmd,
    LPSession session, void*data, size_t size, LPSocketAddr* addr) {
    size_t msgSize = sizeof(LPSocketMsg) + size;
    LPSocketMsg* sm = malloc(sizeof(LPSocketMsg) + size);
    if (data) {
        sm->data = (char*)sm + sizeof(LPSocketMsg);
        memcpy(sm->data, data, size);
    }
    sm->size = size;
    sm->so = s->so;
    if (addr) {
        sm->addr = *addr;
    }
    else {
        sm->addr = s->addr;
    }
    LPSocketAttacher* attacher = s->attacher;
    LPMsgSend send = { 0 };
    send.cmd = cmd;
    send.ptype = LP_PTYPE_SOCKET;
    send.free = freeSocketMsg;
    send.session = session;
    send.data = sm;
    send.size = msgSize;
    lpSend(attacher->attach, &send);
}


static void
pushReadBuff(Socket* s, size_t size, LPSession session) {
    BuffList* bl = malloc(sizeof(BuffList));
    bl->size = size;
    bl->session = session;
    bl->data = NULL;
    bl->next = NULL;
    if (s->rbufEnd) {
        s->rbufEnd->next = bl;
        s->rbufEnd = bl;
    }
    else {
        s->rbuf = bl;
        s->rbufEnd = bl;
    }
}

static BuffList*
popReadBuff(Socket* s) {
    if (s->rbuf) {
        if (s->rbuf == s->rbufEnd) {
            s->rbufEnd = s->rbufEnd->next;
        }
        BuffList* rb = s->rbuf;
        s->rbuf = s->rbuf->next;
        free(rb);
        return s->rbuf;
    }
    return NULL;
}

static void
pushSendBuff(Socket* s, const char* data, size_t size) {
    BuffList* bl = malloc(sizeof(BuffList));
    memset(bl, 0, sizeof(BuffList));
    bl->size = size;
    bl->data = malloc(size);
    bl->next = NULL;
    bl->session = 0;
    memcpy(bl->data, data, size);
    if (s->sbufEnd) {
        s->sbufEnd->next = bl;
        s->sbufEnd = bl;
    }
    else {
        s->sbuf = bl;
        s->sbufEnd = bl;
    }
}

static BuffList*
popSendBuff(Socket* s) {
    BuffList* bl = s->sbuf;
    if (bl) {
        if (bl == s->sbufEnd)
            s->sbufEnd = s->sbufEnd->next;
        s->sbuf = s->sbuf->next;
        return bl->next;
    }
    return NULL;
}

static char*
readCacheBuff(Socket* s, size_t size) {
    s->cbufSize -= size;
    BuffList* bl = s->cbuf;
    BuffList* tmp;
    char* data = malloc(size);
    size_t cur = 0;
    size_t copysize = 0;
    while (bl && size) {
        copysize = (size <= bl->size) ? size : bl->size;
        memcpy(data + cur, bl->data, copysize);
        size -= copysize;
        bl->size -= copysize;
        if (bl->size) {
            memcpy(bl->data, bl->data + copysize, bl->size);
        }
        else {
            tmp = bl;
            bl = bl->next;

            if (s->cbuf == tmp) {
                s->cbuf = bl;
            }
            if (s->cbufEnd == tmp)
                s->cbufEnd = bl;
            free(tmp->data);
            free(tmp);
        }
    }
    return data;
}
static void
pushCacheBuff(Socket* s, char* data, size_t size, int bcopy) {
    BuffList* bl = malloc(sizeof(BuffList));
    if (bcopy) {
        bl->data = malloc(size);
        memcpy(bl->data, data, size);
    }
    else {
        bl->data = data;
    }
    bl->size = size;
    bl->next = NULL;
    bl->session = 0;
    s->cbufSize += size;
    if (s->cbufEnd) {
        s->cbufEnd->next = bl;
        s->cbufEnd = bl;
    }
    else {
        s->cbuf = bl;
        s->cbufEnd = bl;
    }
}

static void
freeBuffList(BuffList* bl) {
    BuffList* tmp;
    while (bl) {
        if (bl->data) {
            free(bl->data);
            bl->data = NULL;
        }
        tmp = bl;
        bl = bl->next;
        free(tmp);
    }
}

static void
freeReadList(Socket* s) {
    BuffList* tmp;
    BuffList* bl = s->rbuf;
    while (bl) {
        respSocketMsg(s,
            LPSOCKET_ERROR, bl->session, NULL, 0, NULL);
        tmp = bl;
        bl = bl->next;
        free(tmp);
    }
}

static void
setNonBlocking(LPSOCK_FD fd) {
#ifdef LP_WINDOWS
    u_long flag = 1;
    ioctlsocket(fd, FIONBIO, &flag);
#else
    int flag = fcntl(fd, F_GETFL, 0);
    if (-1 == flag) {
        return;
    }
    fcntl(fd, F_SETFL, flag | O_NONBLOCK);
#endif
}

static int
getname(LPSocketAddr* u, char* buffer, size_t size) {
    char tmp[INET6_ADDRSTRLEN];
    void* sin_addr = (u->s.sa_family == AF_INET) 
        ? (void*)&u->v4.sin_addr : (void*)&u->v6.sin6_addr;
    int sin_port = ntohs((u->s.sa_family == AF_INET) ? u->v4.sin_port : u->v6.sin6_port);
    if (inet_ntop(u->s.sa_family, sin_addr, tmp, sizeof(tmp))) {
        return snprintf(buffer, size, "%s:%d", tmp, sin_port);
    }
    else {
        buffer[0] = '\0';
        return 0;
    }
}

static void
closeFd(LPSOCK_FD fd) {
#ifdef LP_WINDOWS
    closesocket(fd);
#else
    close(fd);
#endif
}


static void
setKeepalive(LPSOCK_FD fd) {
    int keepalive = 1;
    setsockopt(fd, SOL_SOCKET,
        SO_KEEPALIVE, (void*)&keepalive, sizeof(keepalive));
}


static Socket*
regSocket (LPSocketAttacher* attacher, LPSOCK_FD fd, int status){
    Socket* s;
    LPSocket so = lpRegInStorage(gCtx.storage, &s);
    s->fd = fd;
    s->so = so;
    s->status = status;
    s->attacher = attacher;
    return s;
}

static void
regInPoll(Socket* s, int evt) {
    s->evt = evt;
    lpRegInSocketPoll(s->attacher->poll, s->fd, evt, s);
}

static void
delInPoll(Socket* s) {
    lpDelInSocketPoll(s->attacher->poll, s->fd);
}

static void
shutSocket(Socket* s) {
    delInPoll(s);
    closeFd(s->fd);
    freeBuffList(s->cbuf);
    freeReadList(s);

    switch (s->status) {
    case SS_BIND:
        break;
    case SS_CONNECTING:
        respSocketMsg(s,
            LPSOCKET_ERROR, s->session, NULL, 0, NULL);
        break;
    case SS_CONNECTED:
        if (s->forward) {
            respSocketMsg(s,
                LPSOCKET_ERROR, s->session, NULL, 0, NULL);
        }
        break;
    }
    LPSocketAttacher* attacher = s->attacher;
    lpRecInStorage(gCtx.storage, s->so);
}

static Socket*
findSocket(LPSocketAttacher* attacher, LPSocket so) {
    return lpFindInStorage(gCtx.storage, so);
}


static void
addEvent(Socket* s, int evt) {
    int e = s->evt | evt;
    if (s->evt == e)
        return;
    s->evt = e;
    lpSetEvtInSocketPoll(s->attacher->poll, s->fd, e, s);
}

static void
delEvent(Socket* s, int evt) {
    int e = s->evt & (~evt);
    if (s->evt == e)
        return;
    s->evt = e;
    lpSetEvtInSocketPoll(s->attacher->poll, s->fd, e, s);
}

static LPSOCK_FD
bindAddr(const char* host, uint16_t port,
    int protocol, int family, LPSocketAddr* addr) {
    struct addrinfo hints = { 0 };
    struct addrinfo* list = NULL;
    struct addrinfo* ptr = NULL;
    if (family == LPSOCKET_IPV4) {
        hints.ai_family = AF_INET;
    }
    else if (family == LPSOCKET_IPV6) {
        hints.ai_family = AF_INET6;
    }
    else {
        hints.ai_family = AF_UNSPEC;
    }
    if (protocol == LPSOCKET_TCP) {
        hints.ai_socktype = SOCK_STREAM;
        hints.ai_protocol = IPPROTO_TCP;
    }
    else {
        hints.ai_socktype = SOCK_DGRAM;
        hints.ai_protocol = IPPROTO_UDP;
    }
    int status;
    char ports[16];
    sprintf(ports, "%d", port);
    status = getaddrinfo(host, ports, &hints, &list);
    if (status) {
        printf("getaddrinfo err: %s\n", gai_strerror(status));
        freeaddrinfo(list);
        return 0;
    }
    LPSOCK_FD fd = LPSOCK_INVALID;
    for (ptr = list; ptr != NULL; ptr = ptr->ai_next) {
        fd = socket(ptr->ai_family, ptr->ai_socktype, ptr->ai_protocol);
        if (fd < 0) {
            fd = LPSOCK_INVALID;
            continue;
        }
        setNonBlocking(fd);
        //if (protocol == LPSOCKET_UDP) {
        //    addr->s = *(ptr->ai_addr);
        //    return fd;
        //}
        status = bind(fd, ptr->ai_addr, (int)ptr->ai_addrlen);
        if (status != 0) {
#ifdef LPSYS_WIN
            int wsaerr = WSAGetLastError();
            if (wsaerr != WSAEWOULDBLOCK && wsaerr != WSAEADDRINUSE) {
                closeFd(fd);
                fd = LPSOCK_INVALID;
                continue;
            }
#else
            int e = errno;
            if (e != EINPROGRESS) {
                closeFd(fd);
                fd = LPSOCK_INVALID;
                continue;
            }
#endif
        }
        addr->s = *(ptr->ai_addr);
        break;
    }
    freeaddrinfo(list);
    return fd;
}

static LPSOCK_FD
connectAddr(const char* host, uint16_t port, LPSocketAddr* addr) {
    struct addrinfo hints = { 0 };
    struct addrinfo* list = NULL;
    struct addrinfo* ptr = NULL;
    hints.ai_family = AF_UNSPEC;
    hints.ai_socktype = SOCK_STREAM;
    hints.ai_protocol = IPPROTO_TCP;
    int status;
    char ports[16];
    sprintf(ports, "%d", port);
    status = getaddrinfo(host, ports, &hints, &list);
    if (status) {
        printf("getaddrinfo err: %s\n", gai_strerror(status));
        freeaddrinfo(list);
        return 0;
    }
    LPSOCK_FD fd = LPSOCK_INVALID;
    for (ptr = list; ptr != NULL; ptr = ptr->ai_next) {
        fd = socket(ptr->ai_family, ptr->ai_socktype, ptr->ai_protocol);
        if (fd < 0) {
            fd = LPSOCK_INVALID;
            continue;
        }
        setNonBlocking(fd);
        status = connect(fd, ptr->ai_addr, (int)ptr->ai_addrlen);
        if (status != 0) {
#ifdef LP_WINDOWS
            if (WSAGetLastError() != WSAEWOULDBLOCK) {
                closeFd(fd);
                fd = LPSOCK_INVALID;
                continue;
            }
#else
            int e = errno;
            if (e != EINPROGRESS) {
                closeFd(fd);
                fd = LPSOCK_INVALID;
                continue;
            }
#endif
        }
        addr->s = *(ptr->ai_addr);
        break;
    }
    freeaddrinfo(list);
    return fd;
}

static inline int
sendData(Socket* s, const void* data, size_t size) {
    if (s->protocol == LPSOCKET_UDP) {
        sendto(s->fd, data, (int)size,
            0, &(s->addr.s), sizeof(LPSocketAddr));
        return 1;
    }
    else {
        return send(s->fd, data, (int)size, 0);
    }
    return 0;
}

static int
recvData(Socket* s, char* data, size_t size) {
    int rsize = recv(s->fd, data, (int)size, 0);
    if (rsize <= 0) {
#ifdef LP_WINDOWS
        if ((rsize == 0) || (WSAGetLastError() != WSAEWOULDBLOCK)) {
            shutSocket(s);
            return -1;
        }
#else

        int no = errno;
        if ((rsize == 0) || ((no != EWOULDBLOCK) && (no != EINTR))) {
            shutSocket(s);
            return -1;
        }
#endif
        rsize = 0;
    }
    return rsize;
}


static int
recvTcp(Socket* s, size_t size, LPSession session) {
    LPSocketAttacher* attacher = s->attacher;
    size_t rlen = 0;
    int rsize = -1;
    char* rbuf;
    for (;;) {
        if (s->cbufSize > 0 && s->cbufSize >= size) {
            if (size == 0) {
                size = s->cbufSize;
            }
            rbuf = readCacheBuff(s, size);
            respSocketMsg(s,
                LPSOCKET_CONNECT, session, rbuf, size, NULL);
            free(rbuf);
            return 1;
        }
        if (size == 0) {
            rlen = DEFAULT_RECV_SIZE;
        }
        else {
            rlen = size - s->cbufSize;
        }
        rbuf = attacher->tcpbuffer;
        if (rlen >= DEFAULT_RECV_SIZE) {
            rlen = DEFAULT_RECV_SIZE;
        }
        rsize = recvData(s, rbuf, rlen);
        if (rsize > 0) {
            pushCacheBuff(s, rbuf, rsize, rlen <= DEFAULT_RECV_SIZE);
        }
        else {
            break;
        }
    }
    return rsize;
}



static int
recvUdp(Socket* s, LPSession session) {
    LPSocketAddr addr;
    socklen_t len = sizeof(LPSocketAddr);
    LPSocketAttacher* attacher = s->attacher;
    int size = recvfrom(s->fd, 
        attacher->udpbuffer, MAX_UDP_LEN, 0, &addr.s, &len);
    if (size <= 0) {
        return 0;
    }
    respSocketMsg(s,
        LPSOCKET_READ, session, attacher->udpbuffer, size, &addr);
    return 1;
}

static void
eventInConnecting(LPSocketAttacher* attacher, Socket* s, int evt) {
#if defined(LP_UNIX)
    int error;
    socklen_t len = sizeof(error);
    getsockopt(s->fd, SOL_SOCKET, SO_ERROR, &error, &len);
    if (error) {
        evt = LPSOCKETPOLL_ERROR;
    }
    else {
        evt = evt & (~LPSOCKETPOLL_ERROR);
    }
#endif
    if (evt & LPSOCKETPOLL_ERROR) {
        shutSocket(s);
        return;
    }
    if (evt & LPSOCKETPOLL_WRITE) {
        s->status = SS_CONNECTED;
        s->evt = s->evt | LPSOCKETPOLL_ERROR;
        s->evt = s->evt & (~LPSOCKETPOLL_WRITE);
        lpSetEvtInSocketPoll(s->attacher->poll, s->fd, s->evt, s);
        respSocketMsg(s,
            LPSOCKET_CONNECT, s->session, NULL, 0, NULL);
        s->session = 0;

    }
}

static void
eventInConnectedForward(LPSocketAttacher* attacher, Socket* s, int evt) {
    if (evt & LPSOCKETPOLL_READ) {
        int size;
        for (;;) {
            size = recvData(s, attacher->tcpbuffer, DEFAULT_RECV_SIZE);
            if (size == -1) { break; }
            if (size == 0) { return; }
            respSocketMsg(s, LPSOCKET_READ,
                s->session, attacher->tcpbuffer, size, NULL);
        }
    }
    if (evt & LPSOCKETPOLL_WRITE) {
        BuffList* sb = s->sbuf;
        size_t slen = 0;
        int n;
        while (sb) {
            n = sendData(s, sb->data + slen, sb->size - slen);
            if (n <= 0) {
                sb->data += slen;
                sb->size -= slen;
                break;
            }
            slen += n;
            if (slen >= sb->size) {
                slen = 0;
                sb = popSendBuff(s);
                continue;
            }
        }
        if (!sb) {
            delEvent(s, LPSOCKETPOLL_WRITE);
        }
    }
    if ((evt & LPSOCKET_ERROR) && s->so) {
        shutSocket(s);
    }
}

static void
eventInConnected(LPSocketAttacher* attacher, Socket* s, int evt) {
    if (s->forward) {
        eventInConnectedForward(attacher, s, evt);
        return;
    }
    if (evt & LPSOCKETPOLL_READ) {
        BuffList* rb = s->rbuf;
        if (!s->rbuf) {
            delEvent(s, LPSOCKETPOLL_READ);
            return;
        }
        int ret = -1;
        while (rb) {
            ret = recvTcp(s, rb->size, rb->session);
            if (ret == 1) {
                rb = popReadBuff(s);
                continue;
            }
            else {
                break;
            }
        }
        if (ret <= 0) { return; }
        if (!rb) {
            delEvent(s, LPSOCKETPOLL_READ);
        }
    }

    if (evt & LPSOCKETPOLL_WRITE) {
        BuffList* sb = s->sbuf;
        size_t slen = 0;
        int n;
        while (sb) {
            n = sendData(s, sb->data + slen, sb->size - slen);
            if (n <= 0) {
                sb->data += slen;
                sb->size -= slen;
                break;
            }
            slen += n;
            if (slen >= sb->size) {
                slen = 0;
                sb = popSendBuff(s);
                continue;
            }
        }
        if (!sb) {
            delEvent(s, LPSOCKETPOLL_WRITE);
        }
    }
    if ((evt & LPSOCKETPOLL_ERROR)) {
        shutSocket(s);
    }
}

static void
eventInListenForward(LPSocketAttacher* attacher, Socket* s, int evt) {
    if (evt & LPSOCKETPOLL_READ) {
        if (s->protocol == LPSOCKET_UDP) {
            while (recvUdp(s, s->session)) {
            }
            return;
        }
        LPSOCK_FD nfd;
        Socket* ns;
        LPSocketAddr addr;
        socklen_t len = sizeof(LPSocketAddr);

        for (;;) {
            nfd = accept(s->fd, &(addr.s), &len);
            if (nfd == -1) {
                return;
            }
            setKeepalive(nfd);
            setNonBlocking(nfd);
            ns = regSocket(attacher, nfd, SS_CONNECTED);
            ns->addr = addr;
            regInPoll(ns, LPSOCKETPOLL_ERROR);
            ns->evt = LPSOCKETPOLL_ERROR;
            respSocketMsg(ns,
                LPSOCKET_ACCEPT, s->session, NULL, 0, NULL);
        }
    }

}

static void
eventInListen(LPSocketAttacher* attacher, Socket* s, int evt) {
    if (evt & LPSOCKETPOLL_ERROR) {
        shutSocket(s);
        return;
    }
    if (s->forward) {
        eventInListenForward(attacher, s, evt);
        return;
    }
    if (evt & LPSOCKETPOLL_READ) {
        if (!s->rbuf) {
            delEvent(s, LPSOCKETPOLL_READ);
            return;
        }
        LPSOCK_FD nfd;
        Socket* ns;
        LPSocketAddr addr;
        socklen_t len = sizeof(LPSocketAddr);
        BuffList* rb = s->rbuf;
        while (rb) {
            nfd = accept(s->fd, &(addr.s), &len);
            if (nfd == -1) {
                return;
            }
            setKeepalive(nfd);
            setNonBlocking(nfd);

            ns = regSocket(attacher, nfd, SS_CONNECTED);
            ns->addr = addr;
            regInPoll(ns, LPSOCKETPOLL_ERROR);
            ns->evt = LPSOCKETPOLL_ERROR;
            respSocketMsg(ns,
                LPSOCKET_ACCEPT, rb->session, NULL, 0, NULL);
            rb = popReadBuff(s);
        }
        if (!s->rbuf) {
            delEvent(s, LPSOCKETPOLL_READ);
        }
    }
}



static void
pollEvent(Socket* s, int evt) {
    LPSocketAttacher* attacher = s->attacher;
    switch (s->status) {
    case SS_CONNECTING:
        eventInConnecting(attacher, s, evt);
        break;
    case SS_CONNECTED:
        eventInConnected(attacher, s, evt);
        break;
    case SS_LISTEN:
        eventInListen(attacher, s, evt);
        break;
    default:
        assert(0);
        break;
    }
}

static int
initEnv() {
    static int gInitEnv = 0;
    if (gInitEnv) { return 0;  }
#ifdef LP_WINDOWS
    WSADATA* lpwsaData = (WSADATA*)NULL;
    if (lpwsaData == NULL) {
        WSADATA wsaData;
        lpwsaData = &wsaData;
    }
    WORD wVersionRequested = MAKEWORD(1, 1);
    WSAStartup(wVersionRequested, lpwsaData);
    if (LOBYTE(lpwsaData->wVersion) != 1 || HIBYTE(lpwsaData->wVersion) != 1) {
        WSACleanup();
        return 1;
    }
#else
    struct sigaction action;
    action.sa_handler = handlePipe;
    sigemptyset(&action.sa_mask);
    action.sa_flags = 0;
    sigaction(SIGPIPE, &action, NULL);
#endif
    gInitEnv = 1;
    return 0;
}


LP_API int
lpInitSocketContext (){
    gCtx.storage = lpNewStorage(sizeof(Socket), 8);
    return initEnv();
}

LP_API int
lpExitSocketContext (){
    lpFreeStorage(gCtx.storage);
    return 0;
}

LP_API LPSocketAttacher*
lpNewSocketAttacher (LPService attach){
    LPSocketAttacher* attacher = malloc(sizeof(LPSocketAttacher));
    attacher->poll = lpNewSocketPoll();
    attacher->attach = attach;
    return attacher;
}

LP_API LPSocket
lpBindAddr(LPSocketAttacher* attacher,
    const char * host, uint16_t port, int protocol, int family){
    LPSocketAddr addr;
    LPSOCK_FD fd = bindAddr(host, port, protocol, family, &addr);
    if (fd == LPSOCK_INVALID) { return 0;  }
    Socket* s = regSocket(attacher, fd, SS_BIND);
    s->protocol = protocol;
    s->family = family;
    s->addr = addr;
    return s->so;
}


LP_API int
lpListenSocket(LPSocketAttacher* attacher, LPSocket so, int backlog) {
    Socket* s = findSocket(attacher, so);
    if (!s || s->status != SS_BIND) {
        return 1;
    }
    if (s->protocol == LPSOCKET_TCP) {
        if (listen(s->fd, backlog) == -1) {
            return 1;
        }
    }
    else {
//        if (bind(s->fd, &(s->addr.s), sizeof(LPSocketAddr))) {
//#ifdef LP_WINDOWS
//            printf("Wsaerror: %d\n", WSAGetLastError());
//#else
//            printf("errno: %d \n", errno);
//#endif
//            return 1;
//        }
    }
    s->status = SS_LISTEN;
    regInPoll(s, LPSOCKETPOLL_ERROR);
    return 0;
}


LP_API LPSocket
lpConnectAddr(LPSocketAttacher* attacher,
    const char* host, uint16_t port, LPSession session) {
    LPSocketAddr addr;
    LPSOCK_FD fd = connectAddr(host, port, &addr);
    if (fd == LPSOCK_INVALID) {
        return 0;
    }
    Socket* s = regSocket(attacher, fd, SS_CONNECTING);
    s->addr = addr;
    s->protocol = LPSOCKET_TCP;
    s->family = LPSOCKET_IPV0;
    s->session = session;
    regInPoll(s, LPSOCKETPOLL_WRITE 
        | LPSOCKETPOLL_ERROR | LPSOCKETPOLL_READ);
    return s->so;
}

LP_API int
lpAcceptSocket(LPSocketAttacher* attacher,
            LPSocket so, LPSession session) {
    Socket* s = findSocket(attacher, so);
    if (!s || s->status != SS_LISTEN || s->forward) {
        return 1;
    }
    //pushReadBuff(s, 0, session);
    //addEvent(s, LPSOCKETPOLL_READ | LPSOCKETPOLL_ERROR);
    //return 0;
    if (!s->rbuf) {
        pushReadBuff(s, 0, session);
        addEvent(s, LPSOCKETPOLL_READ | LPSOCKETPOLL_ERROR);
        return 0;
    }
    LPSocketAddr addr;
    socklen_t len = sizeof(LPSocketAddr);
    LPSOCK_FD fd = accept(s->fd, &(addr.s), &len);
    if (fd != -1) {
        setKeepalive(fd);
        setNonBlocking(fd);
        Socket* ns = regSocket(attacher, fd, SS_CONNECTED);
        ns->addr = addr;
        regInPoll(ns, LPSOCKETPOLL_ERROR);
        respSocketMsg(ns,
            LPSOCKET_ACCEPT, session, NULL, 0, NULL);
    }
    else {
        pushReadBuff(s, 0, session);
        addEvent(s, LPSOCKETPOLL_READ);
    }
    return 0;
}

LP_API int
lpReadSocket(LPSocketAttacher* attacher,
    LPSocket so, size_t size, LPSession session) {
    Socket* s = findSocket(attacher, so);
    if (!s || s->status != SS_CONNECTED || s->forward) {
        return 1;
    }
    pushReadBuff(s, size, session);
    addEvent(s, LPSOCKETPOLL_READ);
    return 0;
}

LP_API int
lpSendSocket(LPSocketAttacher* attacher,
    LPSocket so, const char* data, size_t size) {
    Socket* s = findSocket(attacher, so);
    if (!s || s->status != SS_CONNECTED || s->forward) {
        return 1;
    }
    size_t slen = 0;
    if (s->sbuf) {
        int n;
        for (;;) {
            n = sendData(s, data + slen, size - slen);
            if (n <= 0)
                break;
            slen += n;
            if (slen >= size) {
                return 0;
            }
        }
    }
    pushSendBuff(s, data + slen, size - slen);
    addEvent(s, LPSOCKETPOLL_WRITE);
    return 0;
}


LP_API int
lpCloseSocket(LPSocketAttacher* attacher, LPSocket so) {
    Socket* s = findSocket(attacher, so);
    if (!s) {
        return 1;
    }
    shutSocket(s);
    return 0;
}

LP_API int
lpForwardSocket(LPSocketAttacher* attacher,
    LPSocket so, LPSession session) {
    Socket* s = findSocket(attacher, so);
    if (!s || s->status == SS_CONNECTING || s->status == SS_BIND) {
        return 1;
    }
    s->forward = 1;
    s->session = session;
    addEvent(s, LPSOCKETPOLL_READ | LPSOCKETPOLL_ERROR);
    return 0;
}

LP_API int
lpDispatchSocket(LPSocketAttacher* attacher) {
#define MAX_POLL_WAIT 64
    LPSocketEvent evts[MAX_POLL_WAIT] = { 0 };
    LPSocketEvent* e;
    Socket* s;
    int n = lpWaitSocketPollEvt(
            attacher->poll, evts, MAX_POLL_WAIT, 1);
    if (n < 1) {
        return 0;
    }
    for (int i = 0; i < n; i++) {
        e = &(evts[i]);
        s = (Socket*)e->ud;
        pollEvent(s, e->evt);
    }
    return n;
}

